# 作业
# 1、统计所有科目都及格的学生，输出结果【姓名，班级，科目名称，分数】
#
# 2、统计总分超过班级平均分的学生，输出结果【姓名，班级，总分】
#
# 3、统计每个班级总分前十的学生，输出结果【姓名，班级，总分】
#
# 4、统计偏科最严重的前10个学生，输出结果【姓名，班级，科目名称，分数】
from pyspark.sql.session import SparkSession

from pyspark.sql.functions import *
from pyspark.sql.window import Window

spark = SparkSession \
    .builder \
    .appName("demo7_student") \
    .config("spark.sql.shuffle.partitions", 1) \
    .getOrCreate()

# 创建表,读取数据
spark.sql("""
CREATE  TABLE if not exists students_tmp2(
    id string ,
    name string ,
    age string  ,
    gender string  ,
    clazz string 
) 
USING CSV
location 'hdfs://master:9000/data/students'
""")

# 将表转换成DF
# spark.sql("select * from students_tmp2").show()
students_df = spark.table("students_tmp2")
students_df.show()

# 分数表
score_df = spark \
    .read \
    .format("csv") \
    .schema("sid STRING,cid STRING,score DOUBLE") \
    .load("/data/scores")

score_df.show()

# 科目表
subject_df = spark \
    .read \
    .format("csv") \
    .schema("cid STRING,c_name STRING,sum_score DOUBLE") \
    .load("/data/subjects")
subject_df.show()


# 1、统计所有科目都及格的学生，输出结果【姓名，班级，科目名称，分数】
# .hint("broadcast"): 将小表广播，实现map join
score_df \
    .join(subject_df, "cid") \
    .where(col("score") >= col("sum_score") * 0.6) \
    .withColumn("num", count("sid").over(Window.partitionBy("sid"))) \
    .where(col("num") == 6) \
    .alias("a") \
    .hint("broadcast") \
    .join(students_df.alias("b"), col("a.sid") == col("b.id")) \
    .select("name", "clazz", "c_name", "score") \
    .show()

# 2、统计总分超过班级平均分的学生，输出结果【姓名，班级，总分】
# 计算总分
sum_score_df = score_df \
    .groupBy("sid") \
    .agg(sum("score").alias("sum_score"))

# 关联获取学生的信息
student_sum_score_df = sum_score_df \
    .alias("a") \
    .join(students_df.alias("b"), col("a.sid") == col("b.id"))

# 1、先分组聚合再关联回去
# DF会被多次使用，如果不缓存需要计算多次
# 计算每个班级的平均分
clazz_avg_score_df = student_sum_score_df \
    .groupBy("clazz") \
    .agg(avg("sum_score").alias("avg_score"))

# 关联获取班级平均分
student_sum_score_df \
    .join(clazz_avg_score_df.hint("broadcast"), "clazz") \
    .where(col("sum_score") > col("avg_score")) \
    .show()

# 2、使用窗口函数实现（推荐）
# 窗口函数不会再map端进行预聚合，如果数据分布不均，会导致数倾斜
student_sum_score_df \
    .withColumn("avg_score", avg("sum_score").over(Window.partitionBy("clazz"))) \
    .where(col("sum_score") > col("avg_score")) \
    .show()

# 3、统计每个班级总分前十的学生，输出结果【姓名，班级，总分】
student_sum_score_df \
    .withColumn("r", row_number().over(Window.partitionBy("clazz").orderBy(col("sum_score").desc()))) \
    .where(col("r") <= 10) \
    .show()

# 4、统计偏科最严重的前10个学生，输出结果【姓名，班级，科目名称，分数】
# 1、关联科目表拿到科目总分，对分数做归一化
# 2、计算标准差（stddev）
top10_id = score_df \
    .join(subject_df, "cid") \
    .withColumn("one_score", col("score") / col("sum_score")) \
    .groupBy("sid") \
    .agg(stddev("one_score").alias("stddev")) \
    .orderBy(col("stddev").desc()) \
    .limit(10) \
    .select("sid")

# 关联获取学生分数
score_df \
    .join(top10_id.hint("broadcast"), "sid") \
    .show()

# 将DF转换成列表
top10_list = top10_id.collect()
top10_list = [i.sid for i in top10_list]
print(top10_list)
# 使用in 实现
score_df.where(col("sid").isin(top10_list)).show()
# while True:
#     pass
